package edu.xhu.im.listener;

import cn.hutool.json.JSONUtil;
import edu.xhu.im.processor.AbstractMessageProcessor;
import edu.xhu.im.processor.ProcessorFactory;
import edu.xhu.model.common.constants.kafka.group.ChatConsumerGroup;
import edu.xhu.model.common.constants.kafka.topic.ChatTopic;
import edu.xhu.model.im.enums.IMCmdType;
import edu.xhu.model.im.pojos.IMRecvInfo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.DependsOn;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.stereotype.Component;
import org.springframework.messaging.handler.annotation.Header;

import java.util.List;

/**
 * IM处理消息类
 */
@Component
@Slf4j
@DependsOn({"IMServerGroup"})
public class KafkaChatConsumerListener {

    //动态topic,根据不通serverId进行消费(x)
    //单机版了,被你kafka坑烂了
    @KafkaListener(topics = ChatTopic.BILL_CHAT_TOPIC, groupId = ChatConsumerGroup.BILI_CHAT_GROUP)
    public void topic_test(List<String> messages,
                           @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                           @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics) {
        //循环处理消息
        for (int i = 0; i < messages.size(); i++) {
            //0.准备
            String message = messages.get(i);   //消息
            Integer partition = partitions.get(i);  //分区
            String topic = topics.get(i);       //主题
            log.info("消费组：" + ChatConsumerGroup.BILI_CHAT_GROUP
                    + " 消费了主题：" + topic
                    + " 的分区：" + partition
                    + " 中的消息：" + message);
            IMRecvInfo info = JSONUtil.toBean(message, IMRecvInfo.class);
            //1.通过code获取type
            IMCmdType type = IMCmdType.fromCode(info.getCmd());
            //2.直接处理就行了,单机版,无需路由
            AbstractMessageProcessor processor = ProcessorFactory.createProcessor(type);
            processor.process(info);
        }
    }
}
